#include <stdio.h>
#ifndef WIN32
#include <sys/time.h>
#else
#define snprintf _snprintf
#endif
#include <rise/common/exmacros.h>
#include <rise/common/ExceptionTemplate.h>
#include <staff/common/DataObject.h>
#include <staff/common/Operation.h>
#include <staff/common/Value.h>
#include "Message.h"
#include "Client.h"
#include "Publisher.h"

#ifdef WIN32
#define EPOCHFILETIME (116444736000000000i64)
#endif

namespace smq
{
  struct Publisher::PublisherImpl
  {
    std::string sPublisherName;
    Client tClient;

    PublisherImpl():
      nMessageNum(0)
    {
    }

    std::string GenerateId()
    {
      char szTimeId[32];
      snprintf(szTimeId, sizeof(szTimeId), "-%04x-%016llx", ++nMessageNum, GetTicks());

      return "id-" + sPublisherName + szTimeId;
    }

    long long GetTicks()
    {
#ifndef WIN32
      struct timeval stTimeVal;
      gettimeofday(&stTimeVal, NULL);
      return stTimeVal.tv_sec * 1000000ll + stTimeVal.tv_usec;
#else
      FILETIME        tFileTime;
      LARGE_INTEGER   lnTime;
      __int64         n64Tmp;

      GetSystemTimeAsFileTime(&tFileTime);
      lnTime.LowPart  = tFileTime.dwLowDateTime;
      lnTime.HighPart = tFileTime.dwHighDateTime;
      n64Tmp  = lnTime.QuadPart;
      n64Tmp -= EPOCHFILETIME;
      n64Tmp /= 10;

      return n64Tmp;
#endif
    }

    void Invoke(staff::COperation& rOperation)
    {
      rOperation.Request().SetNamespaceUri("http://tempui.org/staff-mq/messages/request");
      tClient.Post(rOperation.Request());
      std::string sResultNodeName = rOperation.Request().GetLocalName() + "Result";
      staff::CDataObject tResponse;
      for(;;)
      {
        tClient.Receive(tResponse, false);
        const std::string& sReceivedNodeName = tResponse.GetLocalName();
        if (sReceivedNodeName == sResultNodeName)
        {
          break;
        }
        else
        if (sReceivedNodeName == "Fault")
        {
          RISE_THROWS(rise::CLogicFormatException, tResponse.GetText());
        }
        else
        {
          rise::LogWarning() << "Unknown server message: " << sReceivedNodeName;
        }
      }
      rOperation.Result() = tResponse;
    }

  private:
    int nMessageNum;
  };

  Publisher::Publisher()
  {
    m_pImpl = new PublisherImpl;
  }

  Publisher::~Publisher()
  {
    if (m_pImpl)
    {
      delete m_pImpl;
      m_pImpl = NULL;
    }
  }

  void Publisher::Init(const std::string& sPublisherName, const std::string& sServerHost /*= "127.0.0.1"*/,
                       unsigned short ushServerPort /*= 8768*/)
  {
    m_pImpl->tClient.Init(sServerHost, ushServerPort);
    m_pImpl->sPublisherName = sPublisherName;

    staff::COperation tOperation("Ident");
    tOperation.AddParameter("sName", sPublisherName);
    tOperation.AddParameter("sType", "publisher");
    m_pImpl->Invoke(tOperation);
  }

  void Publisher::Close()
  {
    m_pImpl->tClient.Close();
  }
  
  void Publisher::CreateQueue(const std::string& sQueueName)
  {
    staff::COperation tOperation("CreateQueue");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);
  }

  void Publisher::RemoveQueue(const std::string& sQueueName)
  {
    staff::COperation tOperation("RemoveQueue");
    tOperation.AddParameter("sQueueName", sQueueName);
    m_pImpl->Invoke(tOperation);
  }

  void Publisher::Publish(const std::string& sQueueName, staff::CDataObject& rMessage)
  {
    staff::COperation tOperation("Publish");
    tOperation.AddParameter("sQueueName", sQueueName);
    tOperation.AddParameter("sId", m_pImpl->GenerateId());
    tOperation.AddParameter("sPublisher", m_pImpl->sPublisherName);
    staff::CDataObject tdoMessage("Message");
    tdoMessage.AppendChild(rMessage);
    tOperation.AddParameter(tdoMessage);
    m_pImpl->Invoke(tOperation);
    rMessage.DetachNode();
  }

}

